package realtime.app.dwd.db;/*
 *@ClassName : dwd_trade_order_detail
 *@Description : $ {description}
 *@Author : NingYu
 *@Date : 2023-03-24 14:{MINUTE}
 */

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import realtime.util.MyKafkaUtil;

public class dwd_trade_order_detail {
    public static void main(String[] args) {
        //
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //todo 1.读取kafka数据(两个主题都读。因为需要获取log中的来源id)
        String groupId = "order_detail_0324_04";
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL((groupId)));
        tableEnv.executeSql(
                "create table topic_log(\n" +
                        "    `common` MAP<String,String>,\n" +
                        "    `ts` BIGINT\n" +
                        ")"
                        + MyKafkaUtil.getKafkaDDL("topic_log",groupId));

        //todo 2.读取topic_db中的order_detail
        Table orderDetailTable = tableEnv.sqlQuery("" +
                "select\n" +
                "    data['id'] id,  \n" +
                "    data['course_id'] course_id, \n" +
                "    data['course_name'] course_name, \n" +
                "    data['order_id'] order_id,  \n" +
                "    data['user_id'] user_id, \n" +
                "    data['origin_amount'] origin_amount, \n" +
                "    data['coupon_reduce'] coupon_reduce,\n" +
                "    data['final_amount'] final_amount,\n" +
                "    data['session_id'] session_id,\n" +
                "    data['create_time'] create_time, \n" +
                "    data['update_time'] update_time,\n" +
                "    ts\n" +
                "from topic_db\n" +
                "where  `database`='gmall' " +
                "    and `table` = 'order_detail'\n" +
                "    and `type` = 'insert'");

        tableEnv.createTemporaryView("order_detail",orderDetailTable);

        //tableEnv.sqlQuery("select * from order_detail").execute().print();

        //todo 3.读取topic_db中的order_info
        Table orderInfoTable = tableEnv.sqlQuery("" +
                "select\n" +
                "    data['id'] id,\n" +
                "    data['user_id'] user_id,\n" +
                "    data['origin_amount'] origin_amount,\n" +
                "    data['coupon_reduce'] coupon_reduce,\n" +
                "    data['final_amount'] final_amount,\n" +
                "    data['order_status'] order_status,\n" +
                "    data['out_trade_no'] out_trade_no,\n" +
                "    data['trade_body'] trade_body,\n" +
                "    data['session_id'] session_id,\n" +
                "    data['province_id'] province_id,\n" +
                "    data['create_time'] create_time,\n" +
                "    data['expire_time'] expire_time,\n" +
                "    data['update_time'] update_time,\n" +
                "    ts\n" +
                "from topic_db\n" +
                "where `table` = 'order_info'\n" +
                "    and `type` = 'insert'");
        tableEnv.createTemporaryView("order_info",orderInfoTable);

        //tableEnv.sqlQuery("select * from order_info").execute().print();

        //todo 4.读取topic_db中的cart_info
        Table cartTable = tableEnv.sqlQuery("" +
                "select\n" +
                "    data['course_id'] course_id,\n" +
                "    data['session_id'] session_id,\n" +
                "    data['user_id'] user_id\n" +
                "from topic_db\n" +
                "where `table` = 'cart_info'\n" +
                "    and `type` = 'insert'");
        tableEnv.createTemporaryView("cart_info",cartTable);

        //todo 5.读取topic_log中的页面id及来源id
        Table logTable = tableEnv.sqlQuery("" +
                "select\n" +
                "    `common`['sid'] sid,\n" +
                "    `common`['sc'] sc,\n" +
                "    `common`['uid'] user_id,\n" +
                "    ts\n" +
                "from topic_log\n" +
                "where `common`['sid'] is not null\n" +
                "    and `common`['sc'] is not null");
        tableEnv.createTemporaryView("log_table",logTable);

        //todo 6.三表关联
//        Table resultTable = tableEnv.sqlQuery("" +
//                        "select\n" +
//                        "    od.id,\n" +
//                        "    od.order_id,\n" +
//                        "    od.user_id,\n" +
//                        "    od.course_id,\n" +
//                        "    od.course_name,\n" +
//                        "    oi.province_id,\n" +
//                        "    date_format(od.create_time,'yyyy-MM-dd') create_date,\n" +
//                        "    od.create_time,\n" +
//                        "    od.session_id,\n" +
//                        "    od.update_time,\n" +
//                        "    oi.expire_time,\n" +
//                        "    od.origin_amount,\n" +
//                        "    od.coupon_reduce,\n" +
//                        "    od.final_amount,\n" +
//                        "    od.ts\n" +
//                        "from order_detail od\n" +
//                        "join order_info oi\n" +
//                        "on od.order_id = oi.id\n"
//                );

        Table resultTable = tableEnv.sqlQuery("" +
                "select\n" +
                "    od.id,\n" +
                "    od.order_id,\n" +
                "    od.user_id,\n" +
                "    od.course_id,\n" +
                "    od.course_name,\n" +
                "    oi.province_id,\n" +
                "    date_format(od.create_time,'yyyy-MM-dd') create_date,\n" +
                "    od.create_time,\n" +
                "    ci.session_id,\n" +
                "    lt.sc,\n" +
                "    od.update_time,\n" +
                "    oi.expire_time,\n" +
                "    od.origin_amount,\n" +
                "    od.coupon_reduce,\n" +
                "    od.final_amount,\n" +
                "    od.ts\n" +
                "from order_detail od\n" +
                "join order_info oi\n" +
                "on od.order_id = oi.id\n" +
                "left join cart_info ci\n"+
                "on od.course_id = ci.course_id\n"+
                "inner join log_table lt\n" +
                "on ci.session_id = lt.sid\n"+
                "and ci.user_id = lt.user_id\n"
                +"where lt.sc is not null\n"
        );
        tableEnv.createTemporaryView("result_table",resultTable);

        //tableEnv.sqlQuery("select * from result_table").execute().print();

        //todo 7.建表并写出
        tableEnv.executeSql(""+
                "create table dwd_trade_order_detail(\n" +
                "    id String,\n" +
                "    order_id String,\n" +
                "    user_id String,\n" +
                "    course_id String,\n" +
                "    course_name String,\n" +
                "    province_id String,\n" +
                "    create_date String,\n" +
                "    create_time String,\n" +
                "    session_id String,\n" +
                "    source_id String,\n" +
                "    update_time String,\n" +
                "    expire_time String,\n" +
                "    origin_amount String,\n" +
                "    coupon_reduce String,\n" +
                "    final_amount String,\n" +
                "    ts BIGINT,\n" +
                "primary key(id) not enforced\n" +
                ")"+MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_detail"));


        //
        tableEnv.executeSql("insert into dwd_trade_order_detail select * from result_table");

    }
}
